package com.credithc.signature.job.mq;

import com.credithc.signature.common.utils.UUIDUtil;
import com.credithc.signature.job.configure.rabbitmq.RabbitConfig;
import com.credithc.signature.job.constant.MQConstant;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;

/**
 * 描述：生产者工具类<br/>
 * 作者：sunyang
 * 创建：2019/5/28 11:05
 * 版本：v1.0.0<br/>
 */
@Slf4j
@Component(MQConstant.RABBITSENDER_NAME)
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public class JobMQSender implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {

    private static final Integer VIPONE = 1;
    private static final Integer VIPTWO = 2;
    private static final Integer VIPTHREE = 3;
    private static final Integer VIPFOUR = 4;
    private static final Integer VIPFIVE = 5;

    @Autowired
    RabbitConfig rabbitConfig;

    @Autowired
    RabbitTemplate rabbitTemplate;

    final String id = UUIDUtil.getUUID();

    private volatile boolean failCallBack = false;
    private volatile boolean occurCallBack = false;

    /**
     * 同步发送消息
     *
     * @param message
     * @return
     */
    public boolean sendSync(Object message) {
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
        try {
            CorrelationData correlationData = new CorrelationData(id);
            // 发生MQ消息
            rabbitTemplate.convertAndSend(rabbitConfig.getExchangeName(), rabbitConfig.getQueueName(), message,
                    messageNoneDurablePostProcessor, correlationData);
            if (!occurCallBack) {
                synchronized (this) {
                    // TODO 消息发送失败后没有重试
                    this.wait(1000);
                }
            }
            return !failCallBack;
        } catch (Exception e) {
            log.error("MQ发送数据失败......", e);
        }
        return false;
    }

    private final MessagePostProcessor messageNoneDurablePostProcessor = new MessagePostProcessor() {
        @Override
        public Message postProcessMessage(final Message message) throws AmqpException {
            message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
            return message;
        }
    };

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        failCallBack = !ack || failCallBack;
        occurCallBack = true;
        synchronized (this) {
            this.notifyAll();
        }
        if (!failCallBack) {
            log.debug("成功发送消息:requestSn={},", correlationData.getId());
        } else {
            log.error("发送消息:requestSn={},失败:cause{}", correlationData.getId(), cause);
        }
    }

    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.error("replyCode={},replyText={},exchange={},routingKey={},消息转换失败:{},", replyCode, replyText, exchange, routingKey, message.toString());
        failCallBack = true;
    }

}
